package com.duowan.realtime.computing;

import java.util.Hashtable;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import com.duowan.realtime.thirft.api.BloomFilterException;
import com.duowan.realtime.thirft.api.BloomFilterGroupQuery;
import com.duowan.realtime.thirft.api.BloomFilterService;
import com.duowan.realtime.thirft.api.BloomFilterService.Client;
import com.duowan.realtime.thirft.api.BloomFilterService.Iface;
import com.duowan.realtime.thirft.api.Constants;
import com.duowan.realtime.thirft.api.DistinctData;
import com.duowan.realtime.thirft.api.DistinctRequest;

public class RawBloomFilterClient implements Iface, InitializingBean,DisposableBean{

	private final static Logger LOG = LoggerFactory.getLogger(BloomFilterClient.class);
	
	private ObjectPool<BloomFilterService.Client> clientPool;
	
	private int port = Constants.DEFAULT_BF_SERVER_PORT;
	private String host;
	private int clientPoolSize = 2;
	
	/**
	 * 默认构造函数
	 */
	public RawBloomFilterClient(){
	}
	
	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	@Override
	public void destroy() throws Exception {
		if(clientPool != null) {
			clientPool.close();
		}
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		if(StringUtils.isBlank(host)) throw new IllegalStateException("host must be not empty");
		if(port <= 0) throw new IllegalArgumentException("port > 0 must be true");
		if(clientPoolSize <= 0) throw new IllegalArgumentException("clientPoolSize > 0 must be true");
		clientPool = new GenericObjectPool<BloomFilterService.Client>(new ClientPoolableObjectFactory(),clientPoolSize);
		LOG.info("init end,server="+host+":"+port+" clientPoolSize:" + clientPoolSize);
	}
	
	@Override
	public Map<String, Integer> notContainsCountAndAdd(String bloomfilterGroup, List<BloomFilterGroupQuery> bfGroupQuery) throws BloomFilterException, TException {
		return doComputing(bloomfilterGroup, bfGroupQuery);
	}

	private Map<String, Integer> doComputing(String bloomfilterGroup, List<BloomFilterGroupQuery> bfGroupQuery) {
		Client client = borrowObject();
		try {
			Map<String, Integer> resultMap = client.notContainsCountAndAdd(bloomfilterGroup, bfGroupQuery);
			returnObject(client);
			return resultMap;
		} catch (BloomFilterException e) {
			returnObject(client);
			throw new RuntimeException("error on run notContainsCountAndAdd method:", e);
		} catch (TException e) {
			invalidateObject(client);
			throw new RuntimeException("error on run notContainsCountAndAdd method:", e);
		}
	}
	
	@Override
	public String ping() throws BloomFilterException, TException {
		Client client = borrowObject();
		try {
			return client.ping();
		}finally {
			returnObject(client);
		}
	}

	private void invalidateObject(Client client) {
		try {
			clientPool.invalidateObject(client);
		} catch (Exception e) {
			throw new RuntimeException("invalidateObject error",e);
		}
	}
	
	private void returnObject(Client client) {
		try {
			clientPool.returnObject(client);
		} catch (Exception e) {
			throw new RuntimeException("returnObject error",e);
		}
	}
	
	private Client borrowObject() {
		try {
			return clientPool.borrowObject();
		} catch (Exception e) {
			throw new RuntimeException("borrowObject error",e);
		}
	}
	
	private class ClientPoolableObjectFactory extends BasePoolableObjectFactory<BloomFilterService.Client> {
		final Map<Client,TTransport> clientTTransportMap = new Hashtable<BloomFilterService.Client, TTransport>();
		@Override
		public BloomFilterService.Client makeObject() throws Exception {
			TTransport transport = new TSocket(host, port);
			TProtocol protocol = new TBinaryProtocol(transport);
			transport.open();
			BloomFilterService.Client client = new BloomFilterService.Client(protocol);
			
			Assert.isTrue(validateObject(client),"client ping() error");
			
			clientTTransportMap.put(client, transport);
			LOG.info("connected_to_server:"+host+":"+port+" clientPool.numActive:"+clientPool.getNumActive()+" clientPool.numIdle:"+clientPool.getNumIdle()+" clientTTransportMap.size:"+clientTTransportMap.size());
			return client;
		}
		
		@Override
		public void destroyObject(BloomFilterService.Client obj) throws Exception {
			TTransport transport = clientTTransportMap.remove(obj);
			if(transport != null) {
				LOG.info("destroyObject() closed_transport, server:"+host+":"+port+" clientPool.numActive:"+clientPool.getNumActive()+" clientPool.numIdle:"+clientPool.getNumIdle()+" clientTTransportMap.size:"+clientTTransportMap.size());
				transport.close();
			}
		}
		
		@Override
		public boolean validateObject(Client obj) {
			String ping = null;
			try {
				ping = obj.ping();
				if(Constants.PING_RESPONSE.equals(ping)) {
					return true;
				}
				return false;
			} catch (Exception e) {
				ping = e.toString();
				return false;
			}finally {
				LOG.info("validateObject,BloomFilterService.Client ping() "+host+" and get response:"+ping);
			}
		}
	}

	@Override
	public Map<String, List<DistinctData>> notContainsAndMark(String bloomfilterGroup, List<DistinctRequest> bfCounterRequest) throws BloomFilterException, TException {
		Client client = borrowObject();
		try {
			Map<String, List<DistinctData>> resultMap = client.notContainsAndMark(bloomfilterGroup, bfCounterRequest);
			returnObject(client);
			return resultMap;
		} catch (BloomFilterException e) {
			returnObject(client);
			throw new RuntimeException("error on run notContainsCountAndAdd method:", e);
		} catch (TException e) {
			invalidateObject(client);
			throw new RuntimeException("error on run notContainsCountAndAdd method:", e);
		}
	}
	
}
